package runner

import (
	"context"
	"errors"
	"fmt"
	"time"

	"gitee.com/flycash/distributed_task_platform/internal/errs"

	"gitee.com/flycash/distributed_task_platform/internal/domain"
	"gitee.com/flycash/distributed_task_platform/internal/event"
	"gitee.com/flycash/distributed_task_platform/internal/service/task"
	"github.com/gotomicro/ego/core/elog"
)

const (
	defaultRetrySleepTime = 500 * time.Millisecond
)

type PlanTaskRunner struct {
	planService task.PlanService
	*NormalTaskRunner
}

func NewPlanRunner(planService task.PlanService, singleRunner *NormalTaskRunner) *PlanTaskRunner {
	return &PlanTaskRunner{
		planService:      planService,
		NormalTaskRunner: singleRunner,
	}
}

func (p *PlanTaskRunner) Run(ctx context.Context, task domain.Task) error {
	// 只需要抢占任务就行
	ta, err := p.acquireTask(ctx, task)
	if err != nil {
		return err
	}
	// 抢占成功，立即创建TaskExecution记录
	exec, err := p.execSvc.Create(ctx, domain.TaskExecution{
		Task:      ta,
		StartTime: time.Now().UnixMilli(),
		Status:    domain.TaskExecutionStatusRunning,
	})
	if err != nil {
		p.releaseTask(ctx, ta)
		return err
	}
	plan, err := p.planService.GetPlan(ctx, task.ID)
	if err != nil {
		return err
	}
	// 找到入口任务
	rootTasks := plan.RootTask()
	for idx := range rootTasks {
		rootTask := rootTasks[idx]
		rootTask.PlanExecID = exec.ID
		go p.run(ctx, rootTask.Task)
	}
	return nil
}

func (p *PlanTaskRunner) acquireTask(ctx context.Context, task domain.Task) (domain.Task, error) {
	acquiredTask, err := p.taskAcquirer.Acquire(ctx, task.ID, task.Version, p.nodeID)
	if err != nil {
		return domain.Task{}, fmt.Errorf("任务抢占失败: %w", err)
	}
	return acquiredTask, nil
}

func (p *PlanTaskRunner) releaseTask(ctx context.Context, task domain.Task) {
	if err := p.taskAcquirer.Release(ctx, task.ID, p.nodeID); err != nil {
		p.logger.Error("释放任务失败",
			elog.Int64("taskID", task.ID),
			elog.String("taskName", task.Name),
			elog.FieldErr(err))
	}
}

func (p *PlanTaskRunner) NextStep(ctx context.Context, task domain.Task) error {
	plan, err := p.planService.GetPlan(ctx, task.PlanID)
	if err != nil {
		return err
	}
	planTask, ok := plan.GetTask(task.Name)
	if !ok {
		return fmt.Errorf("当前任务%s 不属于plan%s", task.Name, plan.Name)
	}
	// 获取后继节点
	tasks := planTask.NextStep()
	if len(tasks) == 0 {
		// 说明没有后继任务,发送plan的结束事件
		err = p.producer.Produce(ctx, event.Event{
			TaskID:         plan.ID,
			Version:        plan.Version,
			ScheduleNodeID: plan.ScheduleNodeID,
			ExecID:         plan.Execution.ID,
			Type:           domain.PlanTaskType,
			ExecStatus:     planTask.TaskExecution.Status,
		})
		if err != nil {
			elog.Error("发送结束plan事件失败", elog.FieldErr(err))
		}
		return err
	}
	//
	for idx := range tasks {
		nextPlanTask := tasks[idx]
		// 所有前驱任务都完成了就可以运行
		nextPlanTask.PlanExecID = plan.Execution.ID
		if nextPlanTask.CheckPre() {
			go p.run(ctx, nextPlanTask.Task)
		}
	}
	return nil
}

// 单个任务的逻辑：不断抢占，直至抢占成功或者被其他节点抢占。
func (p *PlanTaskRunner) run(ctx context.Context, task domain.Task) {
	for {
		err := p.NormalTaskRunner.Run(ctx, task)
		if err != nil && !errors.Is(err, errs.ErrTaskPreemptFailed) {
			p.logger.Error("运行任务失败", elog.FieldErr(err))
			time.Sleep(defaultRetrySleepTime)
			continue
		}
		return
	}
}
